use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result};
use bytes::Bytes;
use http_body_util::{combinators::BoxBody, BodyExt, Full};
use hyper::client::conn::http1::Builder;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response, StatusCode};
use hyper_util::client::legacy::Client;
use hyper_util::rt::TokioIo;
use iroh::{Endpoint, NodeAddr};
use tokio::net::TcpListener;
use tokio::sync::{RwLock, Semaphore};
use tokio::time::timeout;
use tracing::{error, info, warn};

const ALPN: &[u8] = b"dumbpipe/0.1";
const HANDSHAKE: &[u8] = b"dumbpipe-handshake";

// Security constants
const MIN_PORT: u16 = 1024;
const MAX_PORT: u16 = 65535;
const MAX_CONNECTIONS: usize = 100;
const CONNECTION_TIMEOUT: Duration = Duration::from_secs(30);
const REQUEST_TIMEOUT: Duration = Duration::from_secs(60);
// const MAX_BODY_SIZE: usize = 10 * 1024 * 1024; // 10MB - Currently unused
const API_KEY_HEADER: &str = "X-Proxy-API-Key";

/// Web 代理服务器，用于转发 HTTP 请求到本地服务
pub struct WebProxy {
    local_port: u16,
    public_port: u16,
    node_addr: NodeAddr,
    endpoint: Arc<Endpoint>,
    running: Arc<RwLock<bool>>,
    connection_semaphore: Arc<Semaphore>,
    active_connections: Arc<RwLock<std::collections::HashSet<u64>>>,
    api_key: Option<String>,
}

impl WebProxy {
    pub async fn new(
        local_port: u16,
        public_port: u16,
        node_addr: NodeAddr,
        endpoint: Arc<Endpoint>,
        api_key: Option<String>,
    ) -> Result<Self> {
        // Validate ports
        if local_port < MIN_PORT || local_port > MAX_PORT {
            return Err(anyhow::anyhow!(
                "Invalid local port: {}. Must be between {} and {}",
                local_port,
                MIN_PORT,
                MAX_PORT
            ));
        }

        if public_port < MIN_PORT || public_port > MAX_PORT {
            return Err(anyhow::anyhow!(
                "Invalid public port: {}. Must be between {} and {}",
                public_port,
                MIN_PORT,
                MAX_PORT
            ));
        }

        if local_port == public_port {
            return Err(anyhow::anyhow!(
                "Local port and public port cannot be the same: {}",
                local_port
            ));
        }

        Ok(Self {
            local_port,
            public_port,
            node_addr,
            endpoint,
            running: Arc::new(RwLock::new(false)),
            connection_semaphore: Arc::new(Semaphore::new(MAX_CONNECTIONS)),
            active_connections: Arc::new(RwLock::new(std::collections::HashSet::new())),
            api_key,
        })
    }

    pub async fn start(&self) -> Result<()> {
        let running = self.running.read().await;
        if *running {
            return Ok(());
        }

        let listener = TcpListener::bind(&format!("0.0.0.0:{}", self.public_port))
            .await
            .context("Failed to bind to public port")?;

        let local_addr = listener.local_addr()?;
        info!("Web proxy listening on http://{}", local_addr);
        info!("Forwarding to local port {}", self.local_port);

        let endpoint = self.endpoint.clone();
        let node_addr = self.node_addr.clone();
        let running = self.running.clone();
        let local_port = self.local_port;

        {
            let mut running_guard = running.write().await;
            *running_guard = true;
        }

        let running_for_spawn = running.clone();
        let semaphore_for_spawn = self.connection_semaphore.clone();
        let active_connections_for_spawn = self.active_connections.clone();
        let api_key_for_spawn = self.api_key.clone();

        tokio::spawn(async move {
            loop {
                // Check if we should stop
                {
                    let running_guard = running_for_spawn.read().await;
                    if !*running_guard {
                        break;
                    }
                }

                // Accept connection with timeout
                let (stream, remote_addr) =
                    match timeout(CONNECTION_TIMEOUT, listener.accept()).await {
                        Ok(Ok(x)) => x,
                        Ok(Err(err)) => {
                            warn!("Failed to accept connection: {:?}", err);
                            continue;
                        }
                        Err(_) => {
                            warn!("Connection accept timeout");
                            continue;
                        }
                    };

                info!("Accepted web connection from {}", remote_addr);

                // Acquire semaphore permit
                let permit = match semaphore_for_spawn.clone().acquire_owned().await {
                    Ok(permit) => permit,
                    Err(err) => {
                        warn!("Failed to acquire connection permit: {:?}", err);
                        drop(stream);
                        continue;
                    }
                };

                let endpoint = endpoint.clone();
                let node_addr = node_addr.clone();
                let active_connections = active_connections_for_spawn.clone();
                let api_key = api_key_for_spawn.clone();

                // Spawn connection handler
                let handle = tokio::spawn(async move {
                    let result = timeout(
                        REQUEST_TIMEOUT,
                        handle_web_connection(stream, endpoint, node_addr, local_port, api_key),
                    )
                    .await;

                    let _ = permit; // Keep permit alive for the duration of the connection

                    match result {
                        Ok(Ok(())) => {
                            info!("Web connection from {} completed successfully", remote_addr);
                        }
                        Ok(Err(e)) => {
                            warn!("Web connection from {} failed: {:?}", remote_addr, e);
                        }
                        Err(_) => {
                            warn!("Web connection from {} timed out", remote_addr);
                        }
                    }
                });

                // Generate unique connection ID for tracking
                let connection_id = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_nanos() as u64;

                // Track the connection ID for cleanup
                {
                    let mut connections = active_connections.write().await;
                    connections.insert(connection_id);
                }

                // Spawn cleanup task
                let cleanup_connections = active_connections.clone();
                tokio::spawn(async move {
                    let _ = handle.await;
                    let mut connections = cleanup_connections.write().await;
                    connections.remove(&connection_id);
                });
            }
        });

        Ok(())
    }

    pub async fn stop(&self) -> Result<()> {
        let mut running = self.running.write().await;
        *running = false;
        drop(running);

        // Clear connection tracking
        let mut connections = self.active_connections.write().await;
        connections.clear();
        drop(connections);

        info!("Web proxy stopped");
        Ok(())
    }

    // Currently unused method
    /*
    pub fn get_info(&self) -> WebProxyInfo {
        WebProxyInfo {
            local_port: self.local_port,
            public_port: self.public_port,
            node_id: self.node_addr.node_id,
            running: self.running.clone(),
        }
    }
    */
}

/*
#[derive(Clone)]
pub struct WebProxyInfo {
    pub local_port: u16,
    pub public_port: u16,
    pub node_id: iroh::NodeId,
    pub running: Arc<RwLock<bool>>,
}
*/

/// 管理多个 Web 代理实例
pub struct WebProxyManager {
    proxies: HashMap<u16, WebProxy>, // public_port -> WebProxy
    endpoint: Arc<Endpoint>,
}

impl WebProxyManager {
    pub fn new(endpoint: Arc<Endpoint>) -> Self {
        Self {
            proxies: HashMap::new(),
            endpoint,
        }
    }

    pub async fn start_proxy(
        &mut self,
        local_port: u16,
        public_port: u16,
        node_addr: NodeAddr,
        api_key: Option<String>,
    ) -> Result<()> {
        if self.proxies.contains_key(&public_port) {
            return Err(anyhow::anyhow!(
                "Proxy for public port {} already exists",
                public_port
            ));
        }

        let proxy = WebProxy::new(
            local_port,
            public_port,
            node_addr,
            self.endpoint.clone(),
            api_key,
        )
        .await?;
        proxy.start().await?;

        self.proxies.insert(public_port, proxy);
        info!("Started web proxy: {} -> {}", public_port, local_port);
        Ok(())
    }

    pub async fn stop_proxy(&mut self, public_port: u16) -> Result<()> {
        if let Some(proxy) = self.proxies.remove(&public_port) {
            proxy.stop().await?;
            info!("Stopped web proxy on port {}", public_port);
        }
        Ok(())
    }

    pub async fn stop_all_proxies(&mut self) -> Result<()> {
        let ports: Vec<u16> = self.proxies.keys().cloned().collect();
        for port in ports {
            self.stop_proxy(port).await?;
        }
        Ok(())
    }

    /// 获取所有活跃的代理信息用于状态同步
    pub fn get_active_proxies(&self) -> Vec<shared::WebShareProxyInfo> {
        self.proxies
            .values()
            .map(|proxy| shared::WebShareProxyInfo {
                local_port: proxy.local_port,
                public_port: proxy.public_port,
                node_id: proxy.node_addr.node_id,
                status: "active".to_string(), // 可以通过检查running状态来确定
                created_at: std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_secs(),
            })
            .collect()
    }
}

async fn handle_web_connection(
    stream: tokio::net::TcpStream,
    endpoint: Arc<Endpoint>,
    node_addr: NodeAddr,
    local_port: u16,
    api_key: Option<String>,
) -> Result<()> {
    info!("Connecting to node {:?} for web proxy", node_addr);

    let conn = endpoint.connect(node_addr, ALPN).await?;
    info!("Opening bi stream for web proxy");

    let (mut send, recv) = conn.open_bi().await?;
    send.write_all(HANDSHAKE).await?;

    // 使用 QuinnEndpoint 包装流
    let quinn_endpoint = QuinnEndpoint { send, recv };
    let io = TokioIo::new(quinn_endpoint);

    let (_sender, conn) = Builder::new()
        .preserve_header_case(true)
        .title_case_headers(true)
        .handshake::<TokioIo<QuinnEndpoint>, hyper::body::Incoming>(io)
        .await?;

    tokio::spawn(async move {
        if let Err(err) = conn.await {
            warn!("HTTP connection failed: {:?}", err);
        }
    });

    // 创建一个简单的代理服务
    let service = service_fn(move |req: Request<hyper::body::Incoming>| {
        proxy_request(req, local_port, api_key.clone())
    });

    match http1::Builder::new()
        .preserve_header_case(true)
        .title_case_headers(true)
        .serve_connection(TokioIo::new(stream), service)
        .with_upgrades()
        .await
    {
        Ok(_) => info!("Web proxy connection completed successfully"),
        Err(err) => warn!("Web proxy connection failed: {:?}", err),
    }

    Ok(())
}

async fn proxy_request(
    req: Request<hyper::body::Incoming>,
    local_port: u16,
    api_key: Option<String>,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>> {
    // Validate the request method and URI
    let method = req.method();
    let uri = req.uri();

    // Check API key authentication if required
    if let Some(ref required_key) = api_key {
        let provided_key = req
            .headers()
            .get(API_KEY_HEADER)
            .and_then(|value| value.to_str().ok())
            .map(|s| s.to_string());

        match provided_key {
            Some(key) if key == *required_key => {
                // API key is valid
            }
            Some(_) => {
                return Ok(Response::builder()
                    .status(StatusCode::UNAUTHORIZED)
                    .body(full("Invalid API key"))
                    .unwrap()
                    .map(|b| b.boxed()));
            }
            None => {
                return Ok(Response::builder()
                    .status(StatusCode::UNAUTHORIZED)
                    .body(full("API key required"))
                    .unwrap()
                    .map(|b| b.boxed()));
            }
        }
    }

    // Only allow safe methods by default
    match *method {
        hyper::Method::GET
        | hyper::Method::POST
        | hyper::Method::PUT
        | hyper::Method::DELETE
        | hyper::Method::HEAD
        | hyper::Method::OPTIONS => {
            // Safe methods
        }
        _ => {
            return Ok(Response::builder()
                .status(StatusCode::METHOD_NOT_ALLOWED)
                .body(full("Method not allowed"))
                .unwrap()
                .map(|b| b.boxed()));
        }
    }

    let target_url = format!("http://localhost:{}{}", local_port, uri);

    let client =
        Client::builder(hyper_util::rt::TokioExecutor::new()).build_http::<hyper::body::Incoming>();

    // 构建转发请求
    let mut forwarded_req = Request::builder()
        .method(method)
        .uri(&target_url)
        .version(req.version());

    // 安全地复制头部，过滤掉危险头部
    for (name, value) in req.headers() {
        let name_str = name.as_str().to_lowercase();
        match name_str.as_str() {
            // Skip security-sensitive headers
            "authorization" | "cookie" | "set-cookie" | "proxy-authorization" => {
                // Skip these headers for security
            }
            "host" => {
                // Let the client set the Host header automatically
            }
            // Forward other headers
            _ => {
                forwarded_req = forwarded_req.header(name, value);
            }
        }
    }

    // 添加安全头部
    forwarded_req = forwarded_req
        .header("X-Forwarded-For", "127.0.0.1")
        .header("X-Forwarded-Proto", "http")
        .header("X-Proxy-By", "iroh-chat-web-proxy");

    let forwarded_req = forwarded_req.body(req.into_body())?;

    // 发送转发请求，带超时
    let response = match timeout(REQUEST_TIMEOUT, client.request(forwarded_req)).await {
        Ok(Ok(resp)) => resp,
        Ok(Err(e)) => {
            error!("Failed to forward request: {:?}", e);
            return Ok(Response::builder()
                .status(StatusCode::BAD_GATEWAY)
                .body(full("Bad Gateway"))
                .unwrap()
                .map(|b| b.boxed()));
        }
        Err(_) => {
            warn!("Request timeout");
            return Ok(Response::builder()
                .status(StatusCode::GATEWAY_TIMEOUT)
                .body(full("Gateway Timeout"))
                .unwrap()
                .map(|b| b.boxed()));
        }
    };

    Ok(response.map(|b| b.boxed()))
}

fn full<T: Into<Bytes>>(chunk: T) -> BoxBody<Bytes, hyper::Error> {
    Full::new(chunk.into())
        .map_err(|never| match never {})
        .boxed()
}

/// QuinnEndpoint 包装器，实现 AsyncRead 和 AsyncWrite
pub struct QuinnEndpoint {
    pub send: iroh::endpoint::SendStream,
    pub recv: iroh::endpoint::RecvStream,
}

impl tokio::io::AsyncRead for QuinnEndpoint {
    fn poll_read(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf,
    ) -> std::task::Poll<std::io::Result<()>> {
        std::pin::Pin::new(&mut self.recv).poll_read(cx, buf)
    }
}

impl tokio::io::AsyncWrite for QuinnEndpoint {
    fn poll_write(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &[u8],
    ) -> std::task::Poll<Result<usize, std::io::Error>> {
        match std::pin::Pin::new(&mut self.send).poll_write(cx, buf) {
            std::task::Poll::Ready(Ok(n)) => std::task::Poll::Ready(Ok(n)),
            std::task::Poll::Ready(Err(e)) => {
                std::task::Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::Other, e)))
            }
            std::task::Poll::Pending => std::task::Poll::Pending,
        }
    }

    fn poll_flush(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), std::io::Error>> {
        match std::pin::Pin::new(&mut self.send).poll_flush(cx) {
            std::task::Poll::Ready(Ok(())) => std::task::Poll::Ready(Ok(())),
            std::task::Poll::Ready(Err(e)) => {
                std::task::Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::Other, e)))
            }
            std::task::Poll::Pending => std::task::Poll::Pending,
        }
    }

    fn poll_shutdown(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), std::io::Error>> {
        match std::pin::Pin::new(&mut self.send).poll_shutdown(cx) {
            std::task::Poll::Ready(Ok(())) => std::task::Poll::Ready(Ok(())),
            std::task::Poll::Ready(Err(e)) => {
                std::task::Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::Other, e)))
            }
            std::task::Poll::Pending => std::task::Poll::Pending,
        }
    }
}
